Refactor interfaces and engine code #153
Conversation
…per functions to bottom of workflow file
flytekit/common/local_workflow.py
Outdated
| self, | ||
| default_inputs: Dict[str, _promise.Input] = None, | ||
| fixed_inputs: Dict[str, Any] = None, | ||
| schedule=None, |
There was a problem hiding this comment.
can we introduce type hints here?
There was a problem hiding this comment.
added it here, but it's missing from so many places... let's address in future PRs.
| return self._has_registered | ||
|
|
||
|
|
||
| class HasDependencies(object): |
There was a problem hiding this comment.
we are just modeling a graph - albeit poorly
There was a problem hiding this comment.
yeah. dependency tracking in flytekit is confusing. We have nodes, which have upstream nodes, but we also have tasks/workflows/launchplans, which also keep track of upstream entities.
I'd like to get to the point where we only use nodes to keep track of registration order and such. This should make the topological sort easier to reason about as well.
It's not quite just the workflow graph though... if you have a workflow that statically references another workflow, somehow you'll need to know to process the second workflow first. same for default launch plans. we'll probably have to build up a registration graph which contains these inter-entity dependencies. This is effectively what the entity level upstream entity array keeps track of now, but i feel like this is cleaner. anyways, not this pr.
flytekit/common/tasks/task.py
Outdated
| @@ -199,7 +210,14 @@ def fetch_latest(cls, project, domain, name): | |||
| :rtype: SdkTask | |||
| """ | |||
| named_task = _common_model.NamedEntityIdentifier(project, domain, name) | |||
| admin_task = _engine_loader.get_engine().fetch_latest_task(named_task) | |||
| client = _flyte_engine._FlyteClientManager( | |||
There was a problem hiding this comment.
I think this is increasing the blast radius of creating the client manager. I see this code fragment repeated again again and again.
I think
fetch register etc should take in a context that has the client in it. But I understand that is a large change. In the interim have a helper method that creates the client, instead of passing the url and insecure flag repeatedly?
| return execution_data.full_inputs | ||
|
|
||
| if execution_data.inputs.bytes > 0: | ||
| with _common_utils.AutoDeletingTempDir() as t: |
There was a problem hiding this comment.
I actually want to get rid of the concept of AutoDeletingTempDir in the framework code. I think this should be a user decision
|
This is a huge change and I love it! A few comments, mostly to push for next steps. I dont think any of the feedback needs action right away. |
…explaining why it should really be called PythonWorkflow
flytekit/tools/module_loader.py
Outdated
| import importlib | ||
| import pkgutil | ||
|
|
||
| import six | ||
|
|
||
| from flytekit.common.exceptions import user as _user_exceptions | ||
| from flytekit.common.local_workflow import SdkRunnableWorkflow as _PythonWorkflow |
There was a problem hiding this comment.
the _ name is different? Intentional?
| @@ -112,11 +114,13 @@ def workflow(nodes, inputs=None, outputs=None, cls=None, on_failure=None): | |||
| :param dict[Text,Output] outputs: [Optional] A dictionary of output descriptors for a workflow. | |||
| :param T cls: This is the class that will be instantiated from the inputs, outputs, and nodes. This will be used | |||
| by users extending the base Flyte programming model. If set, it must be a subclass of | |||
| :py:class:`flytekit.common.workflow.SdkWorkflow`. | |||
| :py:class:`flytekit.common.local_workflow.PythonWorkflow`. | |||
TL;DR
Changes are:
SdkTask,SdkWorkflow, andSdkLaunchPlanclasses as control plane smart objects.SdkTaskwould directly access the Flyte Admin client tofetcha task instead of going through theFlyteTaskwrapper layer.SdkRunnableWorkflowclass mean to encapsulate the local workflow object (i.e. you only have this object when you have the code locally available, and imported into the Python process). Despite the name, it has no bearing on whether or not a workflow is runnable, either locally or otherwise.Outputwas also moved, but we don't have to if it breaks people's imports.SdkWorkflowhas changed. Hopefully not too many people are using it.RegisterableABC meant to encapsulate control plane capability and aLocalEntityABC meant to encapsulate local-only functionality.futurestuff. We only used stuff that is mandatory in 3.0 and we've decided we're no longer supporting Python 2.Type
Are all requirements met?
Complete description
Please see the above.
Tracking Issue
flyteorg/flyte#226
Follow-up issue
flyteorg/flyte#226